package com.example.springsecurity.hadoop.hive;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.hbase.*;

import org.apache.hadoop.hbase.client.Admin;

import org.apache.hadoop.hbase.client.Connection;

import org.apache.hadoop.hbase.client.ConnectionFactory;

import org.apache.hadoop.hbase.client.Delete;

import org.apache.hadoop.hbase.client.Put;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.client.ResultScanner;

import org.apache.hadoop.hbase.client.Scan;

import org.apache.hadoop.hbase.client.Table;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;

public class HbaseClient {
    public static Configuration configuration;
    public static Connection connection;
    public static Admin admin;

    //建立连接
    public static void init(){
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.rootdir","hdfs://127.0.0.1:9000/hbase");
        try{
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    //关闭连接
    public static void close(){
        try{
            if(admin != null){
                admin.close();
            }
            if(connection != null){
                connection.close();
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    public static void createTable(String tableName,String[] fields) throws IOException {
        init();
        TableName tablename = TableName.valueOf(tableName);
        if(admin.tableExists(tablename)){
            System.out.println(tableName+"表已存在，将删除原表重建新表");
            admin.disableTable(tablename);
            admin.deleteTable(tablename);//删除原来的表
        }
            HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
            for(String str:fields){
                HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(str);
                hTableDescriptor.addFamily(hColumnDescriptor);
            }
            admin.createTable(hTableDescriptor);
            System.out.println("表已创建成功");
        close();
    }

    public static void addRecord(String tableName, String rowKey, String []fields, List<String> values) throws IOException {
        init();
        Table table = connection.getTable(TableName.valueOf(tableName));
        for (int i = 0; i < fields.length; i++) {
          Put put = new Put(rowKey.getBytes());
          String [] cols = fields[i].split(":");
          if(cols.length==1)
          {
          put.addColumn(cols[0].getBytes(), "".getBytes(), values.get(i).getBytes());//因为当输入的是单列族，split仅读出一个字符字符串，即cols仅有一个元素
          }
          else {
          put.addColumn(cols[0].getBytes(), cols[1].getBytes(), values.get(i).getBytes());
          }
          table.put(put);
        }
        table.close();
        close();
    }

    public static void modifyData(String tableName,String rowKey,String column,String value) throws IOException {
     init();
     Table table = connection.getTable(TableName.valueOf(tableName));
     Put put = new Put(rowKey.getBytes());
     String [] cols = column.split(":");

    if(cols.length==1) {
        put.addColumn(column.getBytes(),"".getBytes() , value.getBytes());//qualifier:列族下的列名
    }
    else {
        put.addColumn(cols[0].getBytes(),cols[1].getBytes() , value.getBytes());//qualifier:列族下的列名
    }
    table.put(put);
    table.close();
    close();
    }

    public static void deleteRow(String tableName,String rowKey) throws IOException
    {
     init();
     Table table = connection.getTable(TableName.valueOf(tableName));
     Delete delete = new Delete(rowKey.getBytes());
     table.delete(delete);
     table.close();
     close();
    }

    public static void scanTable(String tableName) throws IOException {
        init();
        Table table = connection.getTable(TableName.valueOf(tableName));
        if(table==null){
            System.out.println(tableName+"表不存在!");
        }
        Scan scan=new Scan();
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println(
                        "行键:" + new String(CellUtil.cloneRow(cell)) + "\t" +
                                "列族:" + new String(CellUtil.cloneFamily(cell)) + "\t" +
                                "列名:" + new String(CellUtil.cloneQualifier(cell)) + "\t" +
                                "值:" + new String(CellUtil.cloneValue(cell)) + "\t" +
                                "时间戳:" + cell.getTimestamp());
            }
        }
        scanner.close();
        table.close();
        close();
    }

    public static void clear() throws IOException {
        String[] fields=new String[] {"data"};
        createTable("t1",fields);
        createTable("t2",fields);
        createTable("t3",fields);
        createTable("t4",fields);
    }

    public static void main(String[] arg) throws IOException {
        String[] fields=new String[] {"data"};
        createTable("t1",fields);
        createTable("t2",fields);
        createTable("t3",fields);
        createTable("t4",fields);
    }
}
